package at.ac.tuwien.swa.sparrow.peer

import ac.at.tuwien.infosys.swa.audio.{SubFingerprint, Fingerprint}
import akka.actor.{PoisonPill, Props, ActorSystem, ActorRef}
import akka.testkit.TestProbe
import at.ac.tuwien.swa.sparrow._
import at.ac.tuwien.swa.sparrow.common._
import at.ac.tuwien.swa.sparrow.conf.SpringTestConfig
import at.ac.tuwien.swa.sparrow.peer.PeerCommand._
import at.ac.tuwien.swa.sparrow.peer.component.{RequestHandler, AudioFingerprintFunction, MusicLibrary}
import at.ac.tuwien.swa.sparrow.server.PeerServiceMock
import at.ac.tuwien.swa.sparrow.server.component.{TicketProvider, PeerManager, UserManager}
import at.ac.tuwien.swa.sparrow.server.persistence.AccountRepository
import com.google.common.net.HostAndPort
import com.sun.net.httpserver.HttpServer
import com.typesafe.config.ConfigValueFactory
import java.util.UUID
import java.util.concurrent.atomic.AtomicLong
import javax.annotation.PostConstruct
import javax.inject.Inject
import org.junit.Assert._
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
import org.junit.{FixMethodOrder, After, Test}
import org.springframework.context.ApplicationContext
import org.springframework.core.io.ClassPathResource
import org.springframework.remoting.httpinvoker.SimpleHttpInvokerServiceExporter
import org.springframework.remoting.support.SimpleHttpServerFactoryBean
import org.springframework.test.annotation.DirtiesContext
import org.springframework.test.annotation.DirtiesContext.ClassMode
import org.springframework.test.context.ContextConfiguration
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner
import scala.Some
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.collection.script.Include
import scala.collection.script.Message
import scala.collection.script.Update
import scala.concurrent.duration._
import scala.reflect.io.File

@ContextConfiguration(classes = Array(classOf[SpringTestConfig]))
@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
@RunWith(classOf[SpringJUnit4ClassRunner])
class PeerActorIT {

	@Inject var ctx: ApplicationContext = _
	@Inject var actorSystem: ActorSystem = _
	@Inject var firstPeer: ActorRef = _
	@Inject var requestHandler: RequestHandler = _
	@Inject var accountRepository: AccountRepository = _
	@Inject var userManager: UserManager = _
	@Inject var musicLibrary: MusicLibrary = _
	@Inject var fingerprintFunction: AudioFingerprintFunction = _
	@Inject var peerManager: PeerManager = _
	@Inject var peerService: PeerServiceMock = _
	@Inject var ticketProvider: TicketProvider = _

	val username = "demo"
	val authToken = AuthToken(1L, UUID.randomUUID.toString)
	var fp: Fingerprint = _

	val firstAddress = HostAndPort fromParts("localhost", DEFAULT_PEER_PORT)
	val secondAddress = HostAndPort fromParts("localhost", DEFAULT_PEER_PORT + 1)
	val secondLib = new MusicLibrary()
	var server: HttpServer = _
	var peerId: UUID = _

	val secondConfig = SpringTestConfig.config
		.withValue("port", ConfigValueFactory.fromAnyRef(DEFAULT_PEER_PORT + 1))
		.withValue("username", ConfigValueFactory.fromAnyRef("demo2"))

	@PostConstruct def postConstruct() {
		accountRepository.save(Account(username, "secret", new AtomicLong(10L)))
		userManager.login(username, authToken)

		fp = fingerprintFunction(new File(new ClassPathResource("music/Estonia.mp3").getFile))
		musicLibrary += (fp -> TrackInfo(null, "Demo Track"))

		ticketProvider.hopCount = 2
	}

	@After def after() {
		peerManager.tickets.clear()
		if (server != null) {
			server.stop(1)
			server = null
		}
	}

	@Test(timeout = 10000) def testRequestLocal() {
		val ticket = requestHandler.request(ticketProvider.create(authToken.id, fp))
		assertEquals(true, ticket.isSuccess)
		while (peerManager.tickets.isEmpty) Thread.sleep(100)
		assertEquals("Demo Track", peerManager.tickets(ticket.get).get.title)
	}

	@Test(timeout = 10000) def testRequestNotFound() {
		val ticket = requestHandler.request(ticketProvider.create(authToken.id, new Fingerprint(0D, 0D, Set.empty[SubFingerprint].asJava)))
		assertEquals(true, ticket.isSuccess)
		while (peerManager.tickets.isEmpty) Thread.sleep(100)
		assertEquals(true, peerManager.tickets(ticket.get).isEmpty)
	}

	@Test(timeout = 30000) def testMultiplePeers() {
		val secondFp = fingerprintFunction apply new File(new ClassPathResource("music/Armenia.mp3").getFile)
		secondLib += (secondFp -> TrackInfo(null, "Armenia"))

		val secondPeer = actorSystem.actorOf(Props(classOf[PeerActor], peerService, secondLib, secondConfig))
		val secondRequestHandler = new RequestHandler(secondPeer, peerService, secondLib)
		server = httpServer(secondRequestHandler)

		peerService.update(Set(PeerInfo(1, secondAddress, System.currentTimeMillis, true)))

		// First attempt
		var ticket = requestHandler.request(ticketProvider.create(authToken.id, secondFp))
		assertEquals(true, ticket.isSuccess)
		while (peerManager.tickets.isEmpty || peerManager.tickets(ticket.get).isEmpty) Thread.sleep(500)
		assertEquals("Armenia", peerManager.tickets(ticket.get).get.title)
		assertEquals(Found, result(ticket.get.id).state)

		// Try again
		peerManager.tickets.clear()
		ticket = requestHandler.request(ticketProvider.create(authToken.id, secondFp))
		assertEquals(true, ticket.isSuccess)
		while (peerManager.tickets.isEmpty || peerManager.tickets(ticket.get).isEmpty) Thread.sleep(500)
		assertEquals("Armenia", peerManager.tickets(ticket.get).get.title)
		assertEquals(Found, result(ticket.get.id).state)
		secondPeer ! PoisonPill
	}

	@Test(timeout = 30000) def testMultiplePeersFail() {
		val secondFp = fingerprintFunction(new File(new ClassPathResource("music/Slovak Republic.mp3").getFile))

		val secondPeer = actorSystem.actorOf(Props(classOf[PeerActor], peerService, secondLib, secondConfig))
		val secondRequestHandler = new RequestHandler(secondPeer, peerService, secondLib)
		server = httpServer(secondRequestHandler)

		val ticket = requestHandler.request(ticketProvider.create(authToken.id, secondFp))
		if (ticket.isFailure) {
			ticket.failed.get.printStackTrace()
		}
		assertEquals(true, ticket.isSuccess)

		val results = ListBuffer[(Ticket, Option[TrackInfo])]()
		peerManager.tickets.subscribe(new mutable.Subscriber[Message[(Ticket, Option[TrackInfo])], mutable.ObservableMap[Ticket, Option[TrackInfo]]]() {
			def notify(pub: mutable.ObservableMap[Ticket, Option[TrackInfo]], event: Message[(Ticket, Option[TrackInfo])]) {
				event match {
					case Include(l, elem) => results += elem
					case Update(l, elem) => results += elem
					case _ =>
				}
			}
		})

		while (results.size < 2)	Thread.sleep(500)
		results foreach(r => assertEquals(true, r._2.isEmpty))

		val r = result(ticket.get.id)
		assertEquals(NotFound, r.state)
		secondPeer ! PoisonPill
	}

	@Test(timeout = 10000) def testUpdatePeerOffline() {
		val secondPeer = actorSystem.actorOf(Props(classOf[PeerActor], peerService, secondLib, secondConfig))

		val probe = TestProbe()(actorSystem)
		probe.send(firstPeer, PeerIdentify)
		val pi1 = probe.expectMsgType[PeerIdentity](10.seconds).peerInfo
		probe.send(secondPeer, PeerIdentify)
		val pi2 = probe.expectMsgType[PeerIdentity](10.seconds).peerInfo

		probe.send(secondPeer, PeerNetwork(Set(pi1, pi2)))
		probe.send(secondPeer, PeerLookup)
		val peerNetwork = probe.expectMsgType[PeerNetwork](10.seconds)
		assertEquals(2, peerNetwork.peers.size)

		// Faking peer1 to be offline
		probe.send(secondPeer, Notification(PeerNetwork(Set(PeerInfo(pi1.id, pi1.address, Long.MaxValue, false)))))
		while (peerManager.onlinePeers.size > 1) {
			Thread sleep 100
		}
		// Terminating peer2
		firstPeer ! PoisonPill
		secondPeer ! PoisonPill
		while (peerManager.onlinePeers.size > 0) {
			Thread sleep 100
		}
	}

	def httpExporter(requestHandler: RequestHandler): com.sun.net.httpserver.HttpHandler = {
		val exporter = new SimpleHttpInvokerServiceExporter()
		exporter.setService(requestHandler)
		exporter.setServiceInterface(classOf[PeerInterface])
		exporter.afterPropertiesSet()
		exporter
	}

	def httpServer(requestHandler: RequestHandler) = {
		val factoryBean = new SimpleHttpServerFactoryBean()
		factoryBean.setPort(at.ac.tuwien.swa.sparrow.DEFAULT_PEER_PORT + 1)
		factoryBean.setContexts(Map("/remoting/PeerInterface" -> httpExporter(requestHandler)).asJava)
		factoryBean.afterPropertiesSet()
		factoryBean.getObject
	}

	def result(ticketId: UUID): TrackResult = {
		val found = peerManager.tickets find (t => t._1.id == ticketId)
		found match {
			case None => throw new IllegalArgumentException()
			case Some((_, title)) if title.isDefined => TrackResult(title.get, Found)
			case Some((ticket, title)) if ticket.isExpired => TrackResult(title getOrElse null, NotFound)
			case Some((_, title)) => TrackResult(title getOrElse null, InProgress)
		}
	}
}
